package com.example.mq.mqserver.datacenter;


import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/*
 * 通过这个类, 来针对硬盘上的消息进行管理
 */
public class MessageFileManager {

    static public void init(){

    }

    //这个内部类用来表示该队列的统计信息
    static public class Stat {
        public int totalCount;//总消息数量
        public int validCount;//有效消息数量
    }

    /*
     * 预定消息文件所在目录和文件名
     * 这个方法获取指定队列对应的消息文件所在目录
     * */
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    //获取指定队列的消息数据的文件路径
    //二进制文件，用txt做后缀不合适，这里就不改了 .bin .data二进制
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //获取该队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    //读取队列统计文件信息，从硬盘中
    private Stat readStat(String queueName) {
        //stat消息统计文件是文本文件，直接用scan即可
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();//消息总数
            stat.validCount = scanner.nextInt();//合法消息数
            scanner.close();
            return stat;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }

    }

    //把队列统计信息，写入硬盘
    private void writeStat(String queueName, Stat stat) {


        //outputStream打开文件默认会把源文件清空，此时相当于新的数据覆盖了旧的，append="true"
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();//清理缓冲区，进入硬盘
            printWriter.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //创建队列对应的文件和目录
    public void createQueueFile(String queueName) throws IOException {
        //1、创建队列的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            //不存在，创建这个队列目录(创建多级目录用mkdirs)
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("【createQueueFile】队列目录创建失败" + baseDir.getAbsolutePath());
            }
        }
        //2、创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建队列数据文件失败：" + queueDataFile.getAbsolutePath());
            }
        }

        //3、创建队列统计数据文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建队列统计数据文件失败：" + queueStatFile.getAbsolutePath());
            }
        }

        //4、给消息统计文件设定初始值0/t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    //删除队列目录和文件
    public void destroyQueueFile(String queueName) throws IOException {
        //先删除里面的文件，然后删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();

        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();

        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();

        if (!(ok1 && ok2 && ok3)) {
            throw new IOException("队列文件删除失败：" + baseDir.getAbsolutePath());
        }
    }

    //检查队列的目录和文件是否存在
    public boolean checkFilesExist(String queueName) {
        //判断队列数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }

        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }

        return true;
    }


    //把新的消息放到队列中对应的文件
    //queue是指定队列 message是需要写入的消息
    public void sendMessage(MSGQueue queue, Message message) throws IOException, MqException {
        //1、检查当前要写入队列的对应文件是否存在
        if (!checkFilesExist(queue.getName())) {
            throw new MqException("发送消息，但是发现队列对应文件不存在，queueName=" + queue.getName());
        }

        //2、把message进行序列化
        byte[] messageBinary = BinaryTool.toBytes(message);

        synchronized (queue) {//保证数据可靠，进行加锁

            //3、获取当前队列数据的数据长度，计算offsetBeg offsetEnd
            //把新的message写道队列文件的末尾，此时message的offsetBeg=当前文件长度+4（先4个字节声明消息长度）
            //offset就是当前文件长度+4+message自身长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
//        通过queueDataFile.length();，获取长度，单位字节
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);

            //4、把消息写入文件(此处写入是追加写，不是覆盖写）
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //1. 先写当前文件长度，占用四个字节
                    //这个方法直接写四个字节
                    dataOutputStream.writeInt(messageBinary.length);

                    //2. 写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }

            //5、更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount++;
            stat.validCount++;
            writeStat(queue.getName(), stat);
        }
    }

    //删除指定队列中的指定消息
    //这里是逻辑删除 把isValid设置为 0
    //1、读取文件的一段数据，反序列化为Message对象
    //2、设置isValid=0
    //3、写会文件
    //message必须包含有效的offsetBeg End有效
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, MqException, ClassNotFoundException {
        synchronized (queue) {//加锁

            //这里随机访问的意思是 不用从头一直读到目标位置，只需要直接跳到目标位置进行读取即可
            try (RandomAccessFile randomAccessFile
                         = new RandomAccessFile(
                    getQueueDataPath(queue.getName()), "rw")//r可以读 w可以写
            ) {
                byte[] srcBinary = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());//seek相当于光标
                //read会尝试读取对多的字节放到srcBinary
                randomAccessFile.read(srcBinary);

                //转成对象isValid=0
                Message srcMessage = (Message) BinaryTool.fromBytes(srcBinary);
                srcMessage.setIsValid((byte) 0x0);

                //重新写入文件
                byte[] curBinary = BinaryTool.toBytes(srcMessage);
                randomAccessFile.seek(message.getOffsetBeg());//read也会移动光标，所以这里还要seek
                randomAccessFile.write(curBinary);
            }

            //更新统计文件:有效消息减少1
            Stat stat = readStat(queue.getName());
            if (stat.validCount > 0) stat.validCount--;

            writeStat(queue.getName(), stat);
        }
    }

    //这个方法，读取文件中一个队列的一连串消息，在程序启动时调用，因此不用加锁
    public LinkedList<Message> loadAllMessageForQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        long currentOffset = 0;
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                while (true) {
                    int messageSize = dataInputStream.readInt();
                    byte[] messageBinary = new byte[messageSize];

                    //read方法尽可能填充messageBinary，返回最终实际填充长度
                    int actualSize = dataInputStream.read(messageBinary);
                    if (actualSize != messageSize) {
                        throw new MqException("[loadAllMessageForQueue] 文件格式错乱,文件名：" + queueName);
                    }
                    Message message = (Message) BinaryTool.fromBytes(messageBinary);
                    if (message.getIsValid() == 0x0) {//无效消息，读取下一个
                        currentOffset += (4 + messageSize);
                        continue;
                    }

                    //设置偏移量 读入消息
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);


                }
            } catch (EOFException e) {
                System.out.println("从硬盘读取队列消息完毕");
            }
        }
        return messages;
    }

    //创建新文件区，把有效数据放进来，然后改名即可
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }


    //检查是否进行垃圾回收
    //消息总数大于2000 并且有效消息小于%50
    public boolean checkGC(String queueName) {
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount < 0.5) {
            return true;
        }
        return false;
    }

    public void gc(MSGQueue queue) throws IOException, MqException, ClassNotFoundException {
        synchronized (queue) {//保证线程安全
            long startTime = System.currentTimeMillis();

            //1、创建新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if (queueDataNewFile.exists()) {//已经存在抛出异常，可能存在数据残留问题
                throw new MqException("【MessageFileManager】queue_data_new文件已经存在,可能存在数据残留问题");
            }

            //没有则创建
            boolean ok=queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("新文件创建失败");
            }

            //2、旧的文件中读取所有有效消息
            LinkedList<Message> messages =loadAllMessageForQueue(queue.getName());

            //3、把消息写入新的文件
            try(FileOutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for(Message message : messages) {
                        //把对象转成字节流
                        byte[] bytes=BinaryTool.toBytes(message);
                        //写入长度
                        dataOutputStream.writeInt(bytes.length);
                        //然后写入本体
                        dataOutputStream.write(bytes);
                    }

                }
            }

            //4、删除旧文件数据
            File queueDataOldFile=new File(getQueueDataPath(queue.getName()));
            ok=queueDataOldFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败! queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }

            //5、更改新文件名字为旧文件名
           ok= queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败! queueDataNewFile=" + queueDataNewFile.getAbsolutePath()
                        + ", queueDataOldFile=" + queueDataOldFile.getAbsolutePath());
            }

            //更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount=messages.size();
            stat.validCount=messages.size();
            writeStat(queue.getName(), stat);

            long endTime = System.currentTimeMillis();
            System.out.println("[MessageFileManager] 垃圾回收执行时间："+(endTime-startTime)+"ms");
        }
    }
}
